
package cfar.distributed.map;

import cfar.actors.Location;
import cfar.actors.Message;
import cfar.actors.Node;
import cfar.actors.distributed.Signal;
import cfar.ff.data.Assocs.Assoc2;
import cfar.ff.data.List;
import cfar.ff.data.Maybe;
import cfar.ff.fun.Curry;
import cfar.ff.fun.F0;
import cfar.ff.fun.F1;
import cfar.ff.fun.F5;
import cfar.ff.fun.Seq;
import cfar.ff.interfaces.Sequence;

import java.io.IOException;
import java.io.Serializable;
import java.net.InetAddress;
import java.util.Iterator;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author badi
 */
public class RMap<S1,S2,A,B> implements F5< Sequence<S1,Integer>
                                          , F1<List<B>, B>
                                          , Sequence<S2,Node>
                                          , F1<A,B>
                                          , Sequence<S1,A>
                                          , Sequence<S1,B>
                                          >
{

    /**********************************************************************/
    public Sequence<S1, B> f( Sequence<S1, Integer> redundancies
                            , F1<List<B>, B>        chooser
                            , Sequence<S2, Node>    nodes
                            , F1<A, B>              f
                            , Sequence<S1, A>       seq         )
    {
        List<SentPackage> packages = gen_packages(redundancies, chooser, nodes, f, seq);
        List<Node> sent_nodes = submit_jobs(nodes, packages);
        return collect_results(sent_nodes,(Sequence<S1, B>) seq.<B>empty());
    }
    /**********************************************************************/

    private List<SentPackage> gen_packages ( Sequence<S1, Integer> redundancies
                                           , F1<List<B>, B>        chooser
                                           , Sequence<S2, Node>    nodes
                                           , F1<A, B>              f
                                           , Sequence<S1, A>       seq          )
    {
        List<SentPackage> ps = List.nil();
        Iterator<Node> ns = Seq.cycle(nodes);
        for(Assoc2<Integer, A> assoc : Seq.zip(redundancies, seq)){
            List<Node> ns2 = List.nil();
            for(int i = 0; i < assoc.$1; i++)
                ns2 = ns2.cons(ns.next());
            SentPackage p = new SentPackage(chooser, f, ns2, assoc.$2, assoc.$1);
            ps = ps.snoc(p);
        }
        return ps;
    }

    private List<Node> submit_jobs(Sequence<S2,Node> avail_nodes, List<SentPackage> ps){
        List<Node> nodes = List.nil();

        Iterator<Node> locs = Seq.cycle(avail_nodes);

        for(SentPackage p : ps){
            Node n = locs.next().copy();
            Message msg = new Message(Signal.WorkF0, p);
            n.tell(msg);
            nodes = nodes.snoc(n);
        }


        return nodes;
    }

    private Sequence<S1,B> collect_results(final Sequence<List,Node> workers, final Sequence<S1,B> empty){
        Sequence<S1,B> res = (Sequence<S1, B>) empty.<B>empty();
        for(Node n : workers){
            B b = ( (Assoc2<Object,B>) n.call().payload ).$2;
            res = res.snoc(b);
        }
        return res;
    }

    private class SentPackage implements Serializable, F0<B> {
        final F1<List<B>, B> chooser;
        final F1<A,B> f;
        final java.util.List<Location> locs;
        final A val;
        final int redundancy;

        public SentPackage( final F1<List<B>, B> chooser
                          , final F1<A,B> f
                          , final List<Node> nodes
                          , final A val
                          , final int redundancy)
        {
            java.util.List<Location> _locs = new java.util.LinkedList();
            for(Node n : nodes){
                _locs.add(n.server_location());
            }

            this.chooser = chooser;
            this.f = f;
            this.locs = _locs;
            this.val = val;
            this.redundancy = redundancy;
        }

        public B call() {
            Iterator<Location> _locs = Seq.cycle(locs);
            List<Node> nodes = List.nil();
            for(int i = 0; i < redundancy; i++){
                try {
                    Assoc2<InetAddress, Integer> loc = _locs.next();
                    Node n = new Node(loc.$1, loc.$2);
                    n.tell(new Message(Signal.WorkF0, Curry.curry(f, val)));
                    nodes = nodes.snoc(n);
                } catch (IOException ex) {
                    Logger.getLogger(RMap.class.getName()).log(Level.SEVERE, null, ex);
                }
            }

            List<B> res = List.nil();
            for(Node n : nodes){
                B b = ((Assoc2<Maybe<String>, B>) n.call().payload).$2;
                res = res.cons(b);
            }

            return chooser.f(res);
        }
    }

}
